package com.bw.sparkstreaming.job4
import org.apache.log4j.{Level, Logger}
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.streaming.{Seconds, StreamingContext}


object WordCountForUpdateStateByKey {
  def main(args: Array[String]): Unit = {
    Logger.getLogger("org").setLevel(Level.ERROR)
    val conf = new SparkConf()
    conf.setMaster("local[2]")
    conf.setAppName("WordCount")
    val ssc = new StreamingContext(conf,Seconds(2))
    //如果是在生产里面，我建议大家设置一个HDFS的目录就可以了。
    //但是我这儿HDFS集群超级慢，所以我就设置一个本地目录。
    ssc.checkpoint("/Users/lixiyuan/Desktop/工程/nxsparkstreaming/src/main/data/checkpoint1")
    val dataStream: DStream[String] = ssc.socketTextStream("localhost",8888)
     val lineDSteram: DStream[String] = dataStream.flatMap(_.split(","))
     val wordAndOneDStream: DStream[(String, Int)] = lineDSteram.map((_,1))

    /**
      *  updateFunc: (Seq[V], Option[S]) => Option[S]
      *  参数一：Seq[V]
      *
      *  hadoop,1
      *  hadoop,1
      *  hadoop,1
      *  分组：
      *  {hadoop,{1,1,1}}   values = {1,1,1}
      * 参数二：Option
      *  这个key的上一次的状态(历史的状态)
      *  hadoop,6
      *
      *  返回值：Option
      *  把这个单词出现多少次当做返回值返回去。
      *  Option:
      *     Some:
      *       有值
      *     None：
      *       空
      *
      *
      */

//      val myfunction=(values: Seq[Int], state: Option[Int]) =>{
//        val currentCount = values.sum
//        val lastCount = state.getOrElse(0)
//        //在scala编程里面，最后一行代码就是返回值。
//        Some(currentCount + lastCount)
//      }

    val result = wordAndOneDStream.updateStateByKey((values: Seq[Int], state: Option[Int]) => {
      val currentCount = values.sum
      val lastCount = state.getOrElse(0)
      //在scala编程里面，最后一行代码就是返回值。
      Some(currentCount + lastCount)
    })

   // val result = wordAndOneDStream.updateStateByKey(myfunction)

    result.print()

    ssc.start()
    ssc.awaitTermination()
    ssc.stop()
  }

}
